

package org.srilankaliving.analytics;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;


public class WallTypeAnalyzer extends Configured implements Tool {


    public static class MapClass extends MapReduceBase
            implements Mapper<LongWritable, Text, Text, Text> {


        Text meterial = new Text();

        private Text word = new Text();

        public void map(LongWritable key, Text value,
                        OutputCollector<Text, Text> output,
                        Reporter reporter) throws IOException {
            String line = value.toString();
            String strTownName = null;
            String[] rowData = line.split(" ");
            strTownName = rowData[0];
            int richPresentage = 0;
            if (rowData.length >= 10) {
                for (int intend = 1; intend <= rowData.length - 10; intend++) {
                    strTownName = strTownName + " " + rowData[intend];
                }
                Text townName = new Text(strTownName);
                int total = Integer.parseInt(rowData[rowData.length - 9].replace(",", ""));
                int brick = Integer.parseInt(rowData[rowData.length - 8].replace(",", ""));
                int cement_block_stone = Integer.parseInt(rowData[rowData.length - 7].replace(",", ""));
                int cabook = Integer.parseInt(rowData[rowData.length - 6].replace(",", ""));
                int pressed_soilbricks = Integer.parseInt(rowData[rowData.length - 5].replace(",", ""));
                int mud = Integer.parseInt(rowData[rowData.length - 4].replace(",", ""));
                int kadjan_palmyrah = Integer.parseInt(rowData[rowData.length - 3].replace(",", ""));
                int plank_metal_sheet = Integer.parseInt(rowData[rowData.length - 2].replace(",", ""));
                int other = Integer.parseInt(rowData[rowData.length - 1].replace(",", ""));
                richPresentage = ((brick + cement_block_stone) * 100 / total);
                IntWritable rich = new IntWritable(richPresentage);
                output.collect(townName, new Text(setRating(rich)));
            }

        }
    }

    public static class Reduce extends MapReduceBase
            implements Reducer<Text, Text, Text, Text> {

        public void reduce(Text key, Iterator<Text> values,
                           OutputCollector<Text, Text> output,
                           Reporter reporter) throws IOException {
            while (values.hasNext()) {

                output.collect(key,values.next());
                break;
            }
        }
    }


    public static String setRating(IntWritable value) {
        String grade = null;
        if (value.get() >= 95) {
            grade = "A";
        } else if (value.get() <= 95 && value.get() >= 90) {
            grade = "B";
        } else if (value.get() <= 90 && value.get() >= 80) {
            grade = "C";
        } else if (value.get() <= 80 && value.get() >= 70) {
            grade = "D";
        } else {
            grade = "E";
        }

        return grade;
    }


    public int run(String[] args) throws Exception {
        JobConf conf = new JobConf(WallTypeAnalyzer.class);
        conf.setJobName("wordcount");

        conf.setOutputKeyClass(Text.class);
        conf.setOutputValueClass(Text.class);
        conf.setOutputKeyComparatorClass(IntComparator.class);
        conf.setMapperClass(MapClass.class);
        conf.setReducerClass(Reduce.class);

        FileInputFormat.setInputPaths(conf, new Path(QOLUtils.WALLTYPE_PATH));
        FileOutputFormat.setOutputPath(conf, new Path(QOLUtils.WALLTYPE_OUT_PATH));

        JobClient.runJob(conf);
        return 0;
    }

    public static class IntComparator extends WritableComparator {

        public IntComparator() {
            super(IntWritable.class);
        }

        @Override
        public int compare(byte[] b1, int s1, int l1,
                           byte[] b2, int s2, int l2) {

            Integer v1 = ByteBuffer.wrap(b1, s1, l1).getInt();
            Integer v2 = ByteBuffer.wrap(b2, s2, l2).getInt();

            return v1.compareTo(v2) * (-1);
        }
    }


    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new WallTypeAnalyzer(), args);
        System.exit(res);
    }


}
